
#include <cassert>
#include <string>
#include <sstream>

#include <anb/common/utils.h>

#include "connection_base.h"
#include "utils.h"
#include "loop_action.h"

namespace anb { namespace net {

connection_base::connection_base(loop::ptr loop, uv_tcp_t* tcp, size_t headlen) : uvo_looped(loop, utils::get_peername(tcp), uvo::type::uvo_connection), m_tcp(tcp) {
	m_peername = utils::get_peername(tcp);
    m_tcp->data = this;
	int err = uv_tcp_nodelay(m_tcp, TRUE);
	if (err) {
		log_error(err, "set nodelay failed");
		assert(false);
	}

	init_buffer(headlen);
}

connection_base::~connection_base() {
	stop();
}

void connection_base::init_buffer(size_t headsize) {
	assert(headsize >= LEN_SIZE);

	char* ptr = (char*)anb::common::utils::malloc(headsize);
	assert(ptr);
	m_head.base = ptr;
	m_head.len = headsize;

	m_body.base = NULL;
	m_body.len = 0;

	m_buf = &m_head;
	m_read = 0;
}

//const size_t connection_base::HEAD_SIZE = 4;

const std::string& connection_base::get_peername() const {
    return m_peername;
}

uv_tcp_t* connection_base::get_tcp() const {
    return m_tcp;
}

int connection_base::start() {
    int err;
    err = uv_read_start((uv_stream_t*)m_tcp, _alloc, _read);
    if (err) {
        log_error(err, "start read error");
        return err;
    }
	log("started");
    return 0;
}

void connection_base::stop() {
	uvo_looped::stop();

	if (m_tcp) {
		loop::close(m_tcp);
		m_tcp = NULL;
	}

	if (m_head.base) {
		anb::common::utils::free(m_head.base);
		m_head.base = NULL;
		m_head.len = 0;
	}
	if (m_body.base) {
		anb::common::utils::free(m_body.base);
		m_body.base = NULL;
		m_body.len = 0;
	}

	for (listeners::const_iterator ite = m_listeners.begin(); ite != m_listeners.end(); ite++) {
		std::shared_ptr<listener> listener = ite->lock();
		if (listener) {
			listener->connection_stopped(this);
		}
	}
	m_listeners.clear();
}

connection_base::listener::~listener() {

}

void connection_base::add_listener(const std::weak_ptr<listener>& listener) {
	m_listeners.push_back(listener);
}

void connection_base::_alloc(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
    connection_base* c = (connection_base*)handle->data;
    c->alloc(suggested_size, buf);
}

void connection_base::_read(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf) {
    connection_base* c = (connection_base*)stream->data;
    c->read(nread);
}

void connection_base::alloc(size_t suggested_size, uv_buf_t* buf) {
	buf->len = m_buf->len - m_read;
	buf->base = m_buf->base + m_read;
	assert(buf->len);
}

void connection_base::read(ssize_t nread) {
	if (nread < 0) {
		log("connection closed by peer");
		stop();
		return;
	}

	m_read += nread;
	if (m_read < m_buf->len) {
		return; // continue read
	}
	assert(m_read == m_buf->len);
	m_read = 0;

	if (m_body.base) {

		handle_body(m_body);

		anb::common::utils::free(m_body.base);
		m_body.base = NULL;
		m_body.len = 0;
		m_buf = &m_head;
	} else {

		uv_buf_t segment;
		segment.base = m_head.base + LEN_SIZE;
		segment.len  = m_head.len  - LEN_SIZE;
		handle_head(segment);

		size_t bodysize
			= (((unsigned char)m_head.base[0]) << 24)
			+ (((unsigned char)m_head.base[1]) << 16)
			+ (((unsigned char)m_head.base[2]) <<  8)
			+ (((unsigned char)m_head.base[3])      );
		if (bodysize <= m_head.len) {
			return; // still read head
		}
		bodysize -= m_head.len;

		char* ptr = (char*)anb::common::utils::malloc(bodysize);
		if (!ptr) {
			log("insufficient memory");
			stop();
			return;
		}
		m_body.len = bodysize;
		m_body.base = ptr;
		m_buf = &m_body;
	}
}

void connection_base::handle_head(const uv_buf_t& segment) {
	std::ostringstream stream;
	stream << "handle_head " << segment.len << " bytes";
	log(stream.str());
}
void connection_base::handle_body(const uv_buf_t& segment) {
	std::ostringstream stream;
	stream << "handle_body " << segment.len << " bytes";
	log(stream.str());
}

void connection_base::send(const uv_buf_t* segments, size_t count)
{
	if (!m_tcp) {
		return;
	}
	int err;
	uv_write_t* req = (uv_write_t*)anb::common::utils::malloc(sizeof(uv_write_t));
	err = uv_write(req, (uv_stream_t*)m_tcp, segments, count, _written);
	if (err) {
		log_error(err, "write error");
		anb::common::utils::free(req);
	}
	//stop();
}

void connection_base::send(write* write)
{
	if (!m_tcp) {
		return;
	}

	int err;
	uv_write_t* req = write->get_write();
	req->data = write;
	err = uv_write(req, (uv_stream_t*)m_tcp, write->get(), write->size(), _written);
	if (err) {
		log_error(err, "write error");
		loop_action::destroy(write);
	}
	//stop();
}

void connection_base::_written(uv_write_t* req, int status) {
	loop_action* action = (loop_action*)req->data;
    loop_action::destroy(action);
	if (status) {
		// TODO report error
	}
}

} }